| 16.深入k8s：Informer使用及其源码分析 | 您所在的位置：网站首页 › k8s informer cache › 16.深入k8s：Informer使用及其源码分析 |
16.深入k8s：Informer使用及其源码分析
|
转载请声明出处哦~，本篇文章发布于luozhiyun的博客：https://www.luozhiyun.com 由于这部分的代码是在client-go 中，所以使用的源码版本是client-go 1.19 这次讲解我用了很一些图，尽可能的把这个模块给描述清楚，如果感觉对你有所帮助不妨发一封邮件激励一下我~ Informer机制 机制设计Informer主要有两个作用： 通过一种叫作 ListAndWatch 的方法，把 APIServer 中的 API 对象缓存在了本地，并负责更新和维护这个缓存。ListAndWatch通过 APIServer 的 LIST API“获取”所有最新版本的 API 对象；然后，再通过 WATCH API 来“监听”所有这些 API 对象的变化； 注册相应的事件，之后如果监听到的事件变化就会调用事件对应的EventHandler，实现回调。Informer运行原理如下：
根据流程图来解释一下Informer中几个组件的作用： Reflector：用于监控指定的k8s资源，当资源发生变化时，触发相应的变更事件，如Added事件、Updated事件、Deleted事件，并将器资源对象放到本地DeltaFIFO Queue中； DeltaFIFO：DeltaFIFO是一个先进先出的队列，可以保存资源对象的操作类型； Indexer：用来存储资源对象并自带索引功能的本地存储，Reflector从DeltaFIFO中将消费出来的资源对象存储至Indexer；Reflector 包会和 apiServer 建立长连接，并使用 ListAndWatch 方法获取并监听某一个资源的变化。List 方法将会获取某个资源的所有实例，Watch 方法则监听资源对象的创建、更新以及删除事件，然后将事件放入到DeltaFIFO Queue中； 然后Informer会不断的从 Delta FIFO Queue 中 pop 增量事件，并根据事件的类型来决定新增、更新或者是删除本地缓存；接着Informer 根据事件类型来触发事先注册好的 Event Handler触发回调函数，然后然后将该事件丢到 Work Queue 这个工作队列中。 实例将到了go-client部分的代码，我们可以直接通过实例来进行上手跑动，Informers Example代码示例如下： package main import ( "flag" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/cache" "k8s.io/client-go/tools/clientcmd" "k8s.io/client-go/util/homedir" "log" "path/filepath" "time" ) func main() { var kubeconfig *string //如果是windows，那么会读取C:\Users\xxx\.kube\config 下面的配置文件 //如果是linux，那么会读取~/.kube/config下面的配置文件 if home := homedir.HomeDir(); home != "" { kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file") } else { kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file") } flag.Parse() config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig) if err != nil { panic(err) } clientset, err := kubernetes.NewForConfig(config) if err != nil { panic(err) } stopCh := make(chan struct{}) defer close(stopCh) //表示每分钟进行一次resync，resync会周期性地执行List操作 sharedInformers := informers.NewSharedInformerFactory(clientset, time.Minute) informer := sharedInformers.Core().V1().Pods().Informer() informer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { mObj := obj.(v1.Object) log.Printf("New Pod Added to Store: %s", mObj.GetName()) }, UpdateFunc: func(oldObj, newObj interface{}) { oObj := oldObj.(v1.Object) nObj := newObj.(v1.Object) log.Printf("%s Pod Updated to %s", oObj.GetName(),nObj.GetName()) }, DeleteFunc: func(obj interface{}) { mObj := obj.(v1.Object) log.Printf("Pod Deleted from Store: %s", mObj.GetName()) }, }) informer.Run(stopCh) }要运行这段代码，需要我们将k8s服务器上的~/.kube代码拷贝到本地，我是win10的机器所以拷贝到C:\Users\xxx\.kube中。 informers.NewSharedInformerFactory会传入两个参数，第1个参数clientset是用于与k8s apiserver交互的客户端，第2个参数是代表每分钟会执行一次resync，resync会周期性执行List将所有资源存放再Informer Store中，如果该参数是0，则禁用resync功能。 通过informer.AddEventHandler函数可以为pod资源添加资源事件回调方法，支持3种资源事件回调方法： AddFunc UpdateFunc DeleteFunc通过名称我们就可以知道是新增、更新、删除时会回调这些方法。 在我们初次执行run方法的时候，可以会将监控的k8s上pod存放到本地，并回调AddFunc方法，如下日志： 2020/10/17 15:13:10 New Pod Added to Store: dns-test 2020/10/17 15:13:10 New Pod Added to Store: web-1 2020/10/17 15:13:10 New Pod Added to Store: fluentd-elasticsearch-nwqph 2020/10/17 15:13:10 New Pod Added to Store: kube-flannel-ds-amd64-bjmt2 2020/10/17 15:13:10 New Pod Added to Store: kubernetes-dashboard-65665f84db-jrw6k 2020/10/17 15:13:10 New Pod Added to Store: mongodb 2020/10/17 15:13:10 New Pod Added to Store: web-0 .... 源码解析 初始化 shared Informer初始化shared Informer初始化的时候会调用到informers.NewSharedInformerFactory进行初始化。 文件位置：informers/factory.go func NewSharedInformerFactory(client kubernetes.Interface, defaultResync time.Duration) SharedInformerFactory { return NewSharedInformerFactoryWithOptions(client, defaultResync) } func NewSharedInformerFactoryWithOptions(client kubernetes.Interface, defaultResync time.Duration, options ...SharedInformerOption) SharedInformerFactory { factory := &sharedInformerFactory{ client: client, namespace: v1.NamespaceAll, defaultResync: defaultResync, informers: make(map[reflect.Type]cache.SharedIndexInformer), startedInformers: make(map[reflect.Type]bool), customResync: make(map[reflect.Type]time.Duration), } // Apply all options for _, opt := range options { factory = opt(factory) } return factory }NewSharedInformerFactory方法最终会调用到NewSharedInformerFactoryWithOptions初始化一个sharedInformerFactory，在初始化的时候会初始化一个informers，用来缓存不同类型的informer。 informer 初始化informer初始化会调用sharedInformerFactory的方法进行初始化，并且可以调用不同资源的Informer。 podInformer := sharedInformers.Core().V1().Pods().Informer() nodeInformer := sharedInformers.Node().V1beta1().RuntimeClasses().Informer()定义不同资源的Informer可以用来监控node或pod。 通过调用Informer方法会根据类型来创建Informer，同一类资源会共享同一个informer。 文件路径：informers/factory.go func (f *podInformer) defaultInformer(client kubernetes.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer { //创建informer return NewFilteredPodInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions) } func (f *podInformer) Informer() cache.SharedIndexInformer { //传入上面定义的defaultInformer方法，用于创建informer return f.factory.InformerFor(&corev1.Pod{}, f.defaultInformer) } func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer { f.lock.Lock() defer f.lock.Unlock() //获取informer类型 informerType := reflect.TypeOf(obj) //查找map缓存,如果存在,那么直接返回 informer, exists := f.informers[informerType] if exists { return informer } //根据类型查找resync的周期 resyncPeriod, exists := f.customResync[informerType] if !exists { resyncPeriod = f.defaultResync } //调用defaultInformer方法创建informer informer = newFunc(f.client, resyncPeriod) f.informers[informerType] = informer return informer }调用InformerFor方法的时候会传入defaultInformer方法用于创建informer。 InformerFor方法里面首先会去sharedInformerFactory的map缓存中根据类型查找对应的informer，如果存在那么直接返回，如果不存在，那么则会调用newFunc方法创建informer，然后设置到informers缓存中。 下面我们看一下NewFilteredPodInformer是如何创建Informer的： 文件位置：informers/core/v1/pod.go func NewFilteredPodInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer { return cache.NewSharedIndexInformer( &cache.ListWatch{ ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { if tweakListOptions != nil { tweakListOptions(&options) } //调用apiserver获取pod列表 return client.CoreV1().Pods(namespace).List(context.TODO(), options) }, WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { if tweakListOptions != nil { tweakListOptions(&options) } //调用apiserver监控pod列表 return client.CoreV1().Pods(namespace).Watch(context.TODO(), options) }, }, &corev1.Pod{}, resyncPeriod, indexers, ) }这里是真正的创建一个informer，并注册了List&Watch的回调函数，list回调函数的api类似下面这样： result = &v1.PodList{} err = c.client.Get(). Namespace(c.ns). Resource("pods"). VersionedParams(&opts, scheme.ParameterCodec). Timeout(timeout). Do(ctx). Into(result)构造Informer通过NewSharedIndexInformer完成： func NewSharedIndexInformer(lw ListerWatcher, exampleObject runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer { realClock := &clock.RealClock{} sharedIndexInformer := &sharedIndexInformer{ processor: &sharedProcessor{clock: realClock}, indexer: NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers), listerWatcher: lw, objectType: exampleObject, resyncCheckPeriod: defaultEventHandlerResyncPeriod, defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod, cacheMutationDetector: NewCacheMutationDetector(fmt.Sprintf("%T", exampleObject)), clock: realClock, } return sharedIndexInformer }sharedIndexInformer里面会创建sharedProcessor，设置List&Watch的回调函数，创建了一个indexer，我们这里看一下NewIndexer是怎么创建indexer的： func NewIndexer(keyFunc KeyFunc, indexers Indexers) Indexer { return &cache{ cacheStorage: NewThreadSafeStore(indexers, Indices{}), keyFunc: keyFunc, } }NewIndexer方法创建了一个cache，它的keyFunc是DeletionHandlingMetaNamespaceKeyFunc，即接受一个object，生成它的namepace/name的字符串。cache里面的数据会存放到cacheStorage中，它是一个threadSafeMap用来存储资源对象并自带索引功能的本地存储。 注册EventHandler事件EventHandler事件的注册是通过informer的AddEventHandler方法进行的。在调用AddEventHandler方法的时候，传入一个cache.ResourceEventHandlerFuncs结构体： 文件位置：tools/cache/shared_informer.go func (s *sharedIndexInformer) AddEventHandler(handler ResourceEventHandler) { s.AddEventHandlerWithResyncPeriod(handler, s.defaultEventHandlerResyncPeriod) } func (s *sharedIndexInformer) AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) { s.startedLock.Lock() defer s.startedLock.Unlock() ... //初始化监听器 listener := newProcessListener(handler, resyncPeriod, determineResyncPeriod(resyncPeriod, s.resyncCheckPeriod), s.clock.Now(), initialBufferSize) //如果informer还没启动，那么直接将监听器加入到processor监听器列表中 if !s.started { s.processor.addListener(listener) return } //如果informer已经启动，那么需要加锁 s.blockDeltas.Lock() defer s.blockDeltas.Unlock() s.processor.addListener(listener) //然后将indexer中缓存的数据写入到listener中 for _, item := range s.indexer.List() { listener.add(addNotification{newObj: item}) } }AddEventHandler方法会调用到AddEventHandlerWithResyncPeriod方法中，然后调用newProcessListener初始化listener。 接着会校验informer是否已经启动，如果没有启动，那么直接将监听器加入到processor监听器列表中并返回；如果informer已经启动，那么需要加锁将监听器加入到processor监听器列表中，然后将indexer中缓存的数据写入到listener中。 需要注意的是listener.add方法会调用processorListener的add方法，这个方法会将数据写入到addCh管道中： func (p *processorListener) add(notification interface{}) { p.addCh |
【本文地址】